package com.ebspp.client.buffer;

import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.thrift.TException;

import com.ebspp.client.placement.IPlacement;
import com.ebspp.client.serverManager.ReadConnection;
import com.ebspp.client.serverManager.ServerManager;
import com.ebspp.global.EBSPPTimer;
import com.ebspp.thrift.DataMsg;
import com.ebspp.thrift.ResourceType;

/**
 * Asynchronous buffer reader
 * 
 * It will first open a connection to the master node on which the data resides.
 * It will provision resources if needed from master.
 * The master might return a server, either itself or a slave node for the provision.
 * The return means that the master has provisioned read resources on the returned server for reader.
 * Reader needs to open a connection to that server if necessary.
 * 
 * The reader should read from the returned server, which are termed candidate in this class.
 * 
 * @author x23pan
 *
 */
public class MvAsyncBufReader implements Runnable
{
	boolean alive = true;
	
	MvAsyncBuffer buf;
	int bufID;
	String server = null;
	Map<String, Integer> canResMap;
	ResourceType resType;
	
	IPlacement placementCB = null;
	
	public MvAsyncBufReader(MvAsyncBuffer buffer, String server)
	{
		this.buf = buffer;
		bufID = buf.getBufID();
		this.server = server;
		canResMap = new HashMap<String, Integer>();
		canResMap.put(server, 0);
		resType = buf.getResType();
		
		
		placementCB = buf.getPlacementCB();
	}
	
	
	@Override
	public void run() 
	{
		while(true)
		{
			try 
			{			
				if(buf.getRQueueSize(server) > 0)
				{
					//decide if it needs more resource based on buffer pressure
					int proNum = calProvision(buf.getRQueueSize(server));
					
					if(proNum != 0)
					{
						//note that the server might not have resources to support the read.
						//Hence, it might refer to another candidate to which its data is replicated
						String newCan = ServerManager.rProvision(server, bufID, proNum);
						if(canResMap.get(newCan) != null)
						{
							canResMap.put(newCan, canResMap.get(newCan)+proNum);
							//the proNum might be negative
							if(canResMap.get(newCan) == 0)
							{
								ServerManager.closeRConn(newCan, bufID);
								canResMap.remove(newCan);
							}
						}
						else
						{
							if(ServerManager.createRConn(newCan, resType, bufID, buf.getID()) == null)
								System.err.println("Cannot create read conn to " + newCan);
							else
								canResMap.put(newCan, proNum);							
						}
					}
					
					
					int readNum = calRAmount();
					String candidate = chooseCandidate(readNum);
					if(candidate == null)
					{
						close("no resources for read");
						return;
					}
					List<DataMsg> data = buf.readRQueue(server, readNum);
					try 
					{
						List<DataMsg> finList = ServerManager.read(data, candidate, bufID);
						if(finList == null)
						{
							close("fail to read from " + server +"->"+ candidate);
							return;
						}
						buf.setRFin(finList);
					}
					catch (TException e) {
						// TODO Auto-generated catch block
						close(e.getMessage() +
										   "\nUnable to read from the server, connection error");
						e.printStackTrace();
						return;
					}
				}
				
				
				//here should be modified to a wait()/notify() based strategy
				Thread.sleep(EBSPPTimer.asyncBuffWriteT);
			} 
			catch (InterruptedException e) {
				// TODO Auto-generated catch block
				System.out.println("Interruption for buffer pool write wait");
				e.printStackTrace();
			}
		}
	}
	
	private int calRAmount()
	{
		return Integer.MAX_VALUE;
	}
	
	/**
	 * calculate how much more resource to provision
	 * @return
	 */
	private int calProvision(int size)
	{
		//naive solution. If none, provision 10, otherwise, provision 0
		Iterator<String> canIt = canResMap.keySet().iterator();
		while(canIt.hasNext())
		{
			String aCan = canIt.next();
			if(canResMap.get(aCan) != 0)
				return 0;
		}
		return 10;
	}
	
	private String chooseCandidate(int size)
	{
		Iterator<String> canIt = canResMap.keySet().iterator();
		while(canIt.hasNext())
		{
			String aCan = canIt.next();
			if(canResMap.get(aCan) != 0)
				return aCan;
		}
		return null;
	}
	
	public void close(String msg)
	{
		System.err.println(msg);
		alive = false;
	}
}
